package cmd

import (
	"context"
	"encoding/json"
	"k8s-events-collector/config"
	"k8s-events-collector/pkg/es"
	"k8s-events-collector/pkg/k8s"
	"time"

	"github.com/spf13/cobra"
	"go.uber.org/zap"
	coreV1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/watch"
)

var (
	confFile string
)

var StartCmd = &cobra.Command{
	Use:   "start",
	Short: "启动 K8S Events Collector",
	Long:  "启动 K8S Events Collector",
	RunE: func(cmd *cobra.Command, args []string) error {

		if err := loadConfig(); err != nil {
			return err
		}

		// 初始化全局变量
		loadGlobal()

		return masterExec()
	},
}

func loadConfig() error {
	return config.LoadConfigFromToml(confFile)
}

func loadGlobal() {
	// 全局日志对象
	zap.NewDevelopmentConfig().Level.SetLevel(zap.DebugLevel)
}

func masterExec() error {
	eventClient, err := k8s.GetKubeClient(config.C.Cluster.Kube)
	if err != nil {
		zap.L().Sugar().Errorf("Failed to get k8s client: %v", err)
		return err
	} else {
		zap.L().Sugar().Infof("Success to get k8s client")
	}
	es, err := es.GetEsClient()
	if err != nil {
		zap.L().Sugar().Errorf("Failed to get es client: %v", err)
		return err
	}
	for {
		events, err := eventClient.List(context.TODO(), metav1.ListOptions{})
		if err != nil {
			zap.L().Sugar().Errorf("Failed to load events: %v", err)
			time.Sleep(time.Second)
			continue
		}
		// Do not write old events.

		resourceVersion := events.ResourceVersion

		watcher, err := eventClient.Watch(
			context.TODO(),
			metav1.ListOptions{
				Watch:           true,
				ResourceVersion: resourceVersion})
		if err != nil {
			zap.L().Sugar().Error("Failed to start watch for new events: %v", err)
			time.Sleep(time.Second)
			continue
		}

		watchChannel := watcher.ResultChan()
		// Inner loop, for update processing.
	inner_loop:
		for {
			select {
			case watchUpdate, ok := <-watchChannel:
				if !ok {
					zap.L().Sugar().Error("Event watch channel closed")
					break inner_loop
				}

				if watchUpdate.Type == watch.Error {
					if status, ok := watchUpdate.Object.(*metav1.Status); ok {
						zap.L().Sugar().Error("Error during watch: %#v", status)
						break inner_loop
					}
					zap.L().Sugar().Error("Received unexpected error", zap.Any("object", watchUpdate.Object))
					break inner_loop
				}

				if event, ok := watchUpdate.Object.(*coreV1.Event); ok {
					switch watchUpdate.Type {
					case watch.Added, watch.Modified:
						data, err := json.Marshal(event)
						if err != nil {
							zap.L().Sugar().Fatal(err)
						}
						zap.L().Sugar().Info("Event received", zap.String("event", string(data)))
						esevent := k8s.NewK8sEvents(*event)
						esevent.Req(es)
					case watch.Deleted:
					default:
						zap.L().Sugar().Warnf("Unknown watchUpdate.Type: %#v", watchUpdate.Type)
					}
				} else {
					zap.L().Sugar().Errorf("Wrong object received: %v", watchUpdate)
				}
			}
		}
	}
}

func init() {
	StartCmd.PersistentFlags().StringVarP(&confFile, "config", "f", "config.toml", "the config file path")
	RootCmd.AddCommand(StartCmd)
}
